/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.wx.mq.mqservice;

import cn.ac.iie.di.datadock.rdata.exchange.client.core.session.receive.REAbstractReceiveMessageHandler;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.ConsumePosition;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.REReceiveSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.REReceiveSessionBuilder;
import cn.ac.iie.di.wx.mq.common.RuntimeEnv;
import java.util.Date;
import org.apache.log4j.Logger;
import cn.ac.iie.di.wx.mq.utils.TimeStampUtil;
import java.util.HashMap;
import java.util.concurrent.LinkedBlockingQueue;

/**
 *
 * @author Administrator
 */
public class Receiver {

    static Logger logger = Logger.getLogger(Receiver.class.getName());
    public String reconnection;
    public String topic;
    public String groupName;
    public int threadNum;
    private TimeStampUtil stampUtil;
    private REConnection conn;
    private LinkedBlockingQueue<HashMap<String, String>> redisQueue;
    public String consum_from_timestamp;

    public Receiver(String reconnection, String topic, String groupName, int threadNum, TimeStampUtil stampUtil, LinkedBlockingQueue<HashMap<String, String>> redisQueue,String consum_from_timestamp) {
        this.reconnection = reconnection;
        this.topic = topic;
        this.groupName = groupName;
        this.threadNum = threadNum;
        this.stampUtil = stampUtil;
        this.redisQueue = redisQueue;
        this.consum_from_timestamp=consum_from_timestamp;
    }


    public REConnection getConn() {
        return conn;
    }

    public void setConn(REConnection conn) {
        this.conn = conn;
    }

    public String getReconnection() {
        return reconnection;
    }

    public void setReconnection(String reconnection) {
        this.reconnection = reconnection;
    }

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public String getGroupName() {
        return groupName;
    }

    public void setGroupName(String groupName) {
        this.groupName = groupName;
    }

    public int getThreadNum() {
        return threadNum;
    }

    public void setThreadNum(int threadNum) {
        this.threadNum = threadNum;
    }

    public TimeStampUtil getStampUtil() {
        return stampUtil;
    }

    public void setStampUtil(TimeStampUtil stampUtil) {
        this.stampUtil = stampUtil;
    }

    public LinkedBlockingQueue<HashMap<String, String>> getRedisQueue() {
        return redisQueue;
    }

    public void setRedisQueue(LinkedBlockingQueue<HashMap<String, String>> redisQueue) {
        this.redisQueue = redisQueue;
    }

    public String getConsum_from_timestamp() {
        return consum_from_timestamp;
    }

    public void setConsum_from_timestamp(String consum_from_timestamp) {
        this.consum_from_timestamp = consum_from_timestamp;
    }

    public void startSession() throws REConnectionException {
        conn = new REConnection(reconnection);
        REReceiveSessionBuilder builder = (REReceiveSessionBuilder) conn.getReceiveSessionBuilder(topic);
        builder.setHandler(new ReceiveHandler(stampUtil,redisQueue));
        builder.setFailureHandler(new REAbstractReceiveMessageHandler<byte[]>() {
            @Override
            public boolean handle(byte[] message) {
                logger.error("formatter fuilure ! size = " + message.length);
                return true;
            }
        });
        builder.setGroupName(groupName);
        //测试用的设置，每次换不同的组名，消费同一组数据
//        builder.setConsumPosition(ConsumePosition.CONSUME_FROM_LAST_OFFSET);
//        builder.setConsumPosition(ConsumePosition.CONSUME_FROM_FIRST_OFFSET);
//        builder.setConsumeTimestamp(new Date().getTime());
        //从指定位置和时间开始消费
        builder.setConsumPosition(ConsumePosition.CONSUME_FROM_TIMESTAMP);
        builder.setConsumeTimestamp(new Date(Long.parseLong(consum_from_timestamp)).getTime());
        builder.setConsumeThreadNum(threadNum);
        REReceiveSession session = (REReceiveSession) builder.build();
        session.start();
        logger.info("session start");
    }
}
